package com.zhaosc.rx.example;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

public class HotObservalExample {

	/**
	 * 无论有没有观察者进行订阅，事件始终都会发生 。
	 * 当 Hot Observable 有多 个订阅者时(多个观察者进行订阅时) , 
	 * HotObservable与订阅者们的关系是一对多的关系，可以与多个订阅者共享信息。
	 * 使用 publish操作符，可以让 ColdObservable转换成 HotObservable，
	 * 它将原先的 Observable 转换成 ConnectableObservable。
	 */
	@Test
	public void testHotObserver() {
		
		ConnectableObservable<Long>  observable = Observable.create(new ObservableOnSubscribe<Long>() {
			@Override
			public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
				Observable.interval(10L, TimeUnit.MILLISECONDS, Schedulers.computation()).take(Integer.MAX_VALUE)
						.subscribe(e::onNext);
			}
		}).observeOn(Schedulers.newThread()).publish();

		observable.connect();
		observable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer1:" + t);
			}
		});
		SleepUtils.sleep(1000L);
		
		observable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer2:" + t);
			}
		});

		SleepUtils.sleep(1000L);
	}
	
	/**
	 * 使用 Subject/Processor 
	 */
	@Test
	public void testSubjectAndProcessor() {
		Observable<Long> observable = Observable.create(new ObservableOnSubscribe<Long>() {
			@Override
			public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
				Observable.interval(10L, TimeUnit.MILLISECONDS, Schedulers.computation()).take(Integer.MAX_VALUE)
						.subscribe(e::onNext);
			}
		}).observeOn(Schedulers.newThread());
		
		// 创建Subject/Processor对象完成
		Subject<Long> subject = PublishSubject.<Long>create().toSerialized() ;
		observable.subscribe(subject) ;
		

        //然后在让下游的订阅通过订阅Subject/Processor对象完成Cold Observable -> Hot Observable的转变
		
		subject.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer1:" + t);
				
			}
		});
		
		
//		subject.subscribe(new Observer<Long>() {
//
//			@Override
//			public void onSubscribe(Disposable d) {
//				System.out.println("observer1 onSubscribe"); 
//				
//			}
//
//			@Override
//			public void onNext(Long t) {
//				System.out.println("observer1:" + t);
//			}
//
//			@Override
//			public void onError(Throwable e) {
//				System.out.println("observer1 onError"); 
//				
//			}
//
//			@Override
//			public void onComplete() {
//				System.out.println("observer1 onComplete"); 
//				
//			}
//		});
		
		SleepUtils.sleep(2000L);
		subject.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer2:" + t);
			}
		});

//		subject.subscribe(new Observer<Long>() {
//
//			@Override
//			public void onSubscribe(Disposable d) {
//				System.out.println("observer2 onSubscribe"); 
//				
//			}
//
//			@Override
//			public void onNext(Long t) {
//				System.out.println("observer2:" + t);
//			}
//
//			@Override
//			public void onError(Throwable e) {
//				System.out.println("observer2 onError"); 
//				
//			}
//
//			@Override
//			public void onComplete() {
//				System.out.println("observer2 onComplete"); 
//				
//			}
//		});
		SleepUtils.sleep(1000L);
		
		
	}
	
}
